package com.netwrok;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;


public class IoEngine implements Runnable {
	private boolean running = false;
	private Selector selector = null;
	private ServerSocketChannel server = null;
	private NetworkEvents net = null;

	public void start(NetworkCfg cfg) throws IOException {
		running = true;
		selector = Selector.open();
		
		server = ServerSocketChannel.open();
		server.configureBlocking(false);
		server.socket().bind(new InetSocketAddress(cfg.port));
		server.register(selector, SelectionKey.OP_ACCEPT);
		
		Thread io = new Thread(this);
		io.start();
	}

	public void setNetwork(NetworkEvents net) {
		this.net = net;
	}

	void finish() {
		running = false;
	}


	private int write(SocketChannel channel, ByteBuffer buffers)
			throws IOException {
		return channel.write(buffers);
	}

	private int read(SocketChannel channel, ByteBuffer buffer)
			throws IOException {
		return channel.read(buffer);
	}

	@Override
	public void run() {
		IoQueues queues=IoQueues.getInstance();
		while (running) {
			try {



				selector.select();
				Iterator<SelectionKey> iterator = selector.selectedKeys()
						.iterator();
				while (iterator.hasNext()) {
					SelectionKey key = iterator.next();
					iterator.remove();

					if (!key.isValid()) {
						continue;
					}

					if (key.isAcceptable()) {
						ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
						SocketChannel channel=serverSocketChannel.accept();
						channel.configureBlocking(false);
						IoSession session = new IoSession(channel, this.net);
						channel.register(selector, SelectionKey.OP_READ,
								session);

						queues.addNewSession(session,net);
						queues.registSession4Response(session);
						continue;
					}
					
					SocketChannel channel = (SocketChannel) key.channel();
					IoSession session = (IoSession) key.attachment();
					if (key.isReadable()) {
						
						try {
							int read =read(channel, session.buffer);
							if (read==-1||read==0) {
								session.close();
								queues.addCloseSession(session,net);
								key.cancel();
								continue;
							}
//							if (read==0) {
//								session.close();
//								queues.addCloseSession(session,net);
//								key.cancel();
//								continue;
//							}
							session.buffer.flip();
							net.decode(session);
						} catch (IOException e) {
							e.printStackTrace();
						}

					}
					ConcurrentLinkedQueue<ByteBuffer> buffers=queues.getResponse(session);
					
					while(true){
						ByteBuffer buffer=buffers.peek();
						if(buffer==null){break;}
						int remain=buffer.remaining();
						try {
							int sent=write(channel, buffer);
							if (sent<remain) {
								break;
							}
						} catch (Exception e) {
							queues.addCloseSession(session,net);
							key.cancel();
							break;
						}
						buffers.poll();
					}
					
				}
			} catch (Exception e) {
				e.printStackTrace();
			}

		}
	}
}
